package cn.zyx.config.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * canal实时监听数据库的变化
 * SpringBoot->>CanalClient: 启动监听线程
 * CanalClient->>CanalServer: 建立连接
 * CanalServer->>MySQL: 订阅 Binlog
 * MySQL->>CanalServer: 推送 Binlog 事件
 * CanalServer->>CanalClient: 转发变更数据
 * CanalClient->>SpringBoot: 处理业务逻辑（更新缓存/发送消息）
 *
 * ApplicationRunner ： 在程序启动完成后执行特定操作
 */
@Slf4j
@Component
@ConditionalOnProperty(name = "canal.enable", havingValue = "ture")
public class CanalClient implements ApplicationRunner {

    @Autowired
    private CanalConfig canalConfig;

    @Autowired
    private CanalSubscribeConfig canalSubscribeConfig;

    @Override
    public void run(ApplicationArguments args) {
        new Thread(this::listenToCanal).start();
    }

    private void listenToCanal() {
        // 获取连接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalConfig.getHost(), canalConfig.getPort()),
                canalConfig.getDestination(),
                canalConfig.getUsername(),
                canalConfig.getPassword()
        );

        try {
            connector.connect();
            connector.subscribe(canalSubscribeConfig.getFilter()); // 处理数据库的哪些表
            connector.rollback();
            // while (true) 永久监听数据库
            while (true) {
                Message message = connector.getWithoutAck(canalSubscribeConfig.getBatchSize()); // 最大的批处理数量
                // 每秒监听一次数据库（消息不存在睡眠一秒）
                long batchId = message.getId();
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    Thread.sleep(1000);
                    continue;
                }
                // 处理消息
                processEntries(message.getEntries());
                connector.ack(batchId); // 确认消息
            }
        } catch (Exception e) {
            log.error("Canal 监听异常", e);
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 处理消息
     * @param entries
     */
    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            // 数据类型等于数据库行数据才进行处理
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                // 获取变更，并进行相应处理
                CanalEntry.RowChange rowChange = parseRowChange(entry.getStoreValue());
                //CanalEntry.EventType eventType = rowChange.getEventType();
                //String tableName = entry.getHeader().getTableName();
                //String schemaName = entry.getHeader().getSchemaName();
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    // 处理变更数据
                    handleRowChange(entry.getHeader().getSchemaName(),
                            entry.getHeader().getTableName(),
                            rowChange.getEventType(),
                            rowData);
                }
            }
        }
    }

    /**
     * 解析行变化
     * @param data
     * @return
     */
    private CanalEntry.RowChange parseRowChange(ByteString data) {
        try {
            return CanalEntry.RowChange.parseFrom(data);
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException("解析 RowChange 失败", e);
        }
    }

    /**
     * 处理行数据
     * @param schema
     * @param table
     * @param eventType
     * @param rowData
     */
    private void handleRowChange(String schema,
                                 String table,
                                 CanalEntry.EventType eventType,
                                 CanalEntry.RowData rowData) {
        // 更新前数据
        Map<String, Object> before = parseColumnsToMap(rowData.getBeforeColumnsList());
        // 更新后数据
        Map<String, Object> after = parseColumnsToMap(rowData.getAfterColumnsList());

        String operation = eventType.name();
        log.info("监听到变更 - 数据库: {}, 表: {}, 操作: {}", schema, table, operation);
        log.info("变更前数据: {}", before);
        log.info("变更后数据: {}", after);

        // 根据业务需求处理数据（如更新缓存、发送消息到 MQ）
        // Example: if ("user".equals(table)) updateCache(after);
    }

    private Map<String, Object> parseColumnsToMap(List<CanalEntry.Column> columns) {
        return columns.stream()
                .collect(Collectors.toMap(
                        CanalEntry.Column::getName,
                        column -> column.getIsNull() ? null : column.getValue()
                ));
    }
}
